package com.xxx.exchange;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * Rabbit MQ使用场景三：发布/订阅——广播模式（每个消费者都能接收到消息）
 * 同一个消息被多个消费者获取。一个消费者队列可以有多个消费者实例，只有其中一个消费者实例会消费到消息。
 * 
 * @author guweiqiang
 * 2018年6月13日
 */
public class Subscribler1 {
	
	private static final String EXCHANGE_NAME = "test_exchanger";
	
//	private static final String HOST = "localhost";
	
	public static void main(String[] args) throws IOException, TimeoutException {
		// 1.创建连接工厂，并设置工厂相关属性
		ConnectionFactory factory = new ConnectionFactory();
//		factory.setHost(HOST); // 默认为localhost
		
		// 2.创建一个连接
		Connection connection = factory.newConnection();
		
		// 3.创建一个通道
		Channel channel = connection.createChannel();
		
		// 4.声明一个交换机exchanger
		channel.exchangeDeclare(EXCHANGE_NAME, "fanout");   // type=fanout，subscibler和publisher必须要一致
		// fanout表示分发，所有的消费者得到同样的队列信息
		// RabbitMQ提供了四种Exchange: fanout,direct,topic,header
		// 性能排序：fanout > direct >> topic。比例大约为11：10：6
		
		// 5.产生一个随机的队列名称，然后将队列绑定到exchanger上
		String queueName = channel.queueDeclare().getQueue(); // 每个消费者都是独立的队列
		channel.queueBind(queueName, EXCHANGE_NAME, ""); // 必须将队列绑定到交换机上，否则接收不到消息（routingKey必须为空）
		channel.basicQos(1); // 同一时刻服务器只会发一条消息给消费者
		
		// 6.从队列里接收消息
		System.out.println("Subscribler1 waiting received messages：");
		// DefaultConsumer表示consumer采取的是pull模式来拉取消息
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, 
										Envelope envelope,
										AMQP.BasicProperties properties,
										byte[] body) throws IOException {
				String message = new String(body, "UTF-8");
				System.out.println("Subscribler1 received message：" + message);
				channel.basicAck(envelope.getDeliveryTag(), false); // false表示使用手动确认模式
			}
		};
		
		// 7.消费消息之后回复生产者（批量消费消息后1次性回复）
		channel.basicConsume(queueName, false, consumer); // 监听队列，false手动返回完成
	}
}
